package org.ws.task.item;

import com.google.common.collect.Iterators;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.model.TaskProcessStatus;
import org.ws.task.model.TaskItem;
import org.ws.task.model.TaskStatus;

import java.io.Serializable;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ExecuteContext<T> extends AbstractQueue<TaskItem<T>> implements Serializable {

    private static Logger log = LoggerFactory.getLogger(ExecuteContext.class.getName());

    private static final long serialVersionUID = 5100233584421007903L;
    // 失败重试次数
    private static final int MAX_FAILED_RETRY_COUNT = 3;

    private final Queue<TaskItem<T>> awaitQueue = Queues.newConcurrentLinkedQueue();
    private final AtomicInteger awaitQueueSize = new AtomicInteger(0);

    private final Map<Long,TaskItem<T>> processItemMap = Maps.newConcurrentMap();
    private final AtomicInteger processItemSize = new AtomicInteger(0);

    private final Queue<TaskItem<T>> failedQueue = Queues.newConcurrentLinkedQueue();
    private final AtomicInteger failedQueueSize = new AtomicInteger(0);

    private final Queue<TaskItem<T>> commitFailedQueue = Queues.newConcurrentLinkedQueue();

    private transient final ReadWriteLock lock = new ReentrantReadWriteLock();

    private static final String TRY_LOCK_TIMEOUT = "try lock timeout:";

    @Override
    public Iterator<TaskItem<T>> iterator() {
        try {
            lock.readLock().lockInterruptibly();
            Iterator<TaskItem<T>> iterator = awaitQueue.iterator();
            Iterator<TaskItem<T>> iterator1 = failedQueue.iterator();

            return Iterators.concat(iterator,iterator1);
        }
        catch (Exception e){
            log.error(TRY_LOCK_TIMEOUT+e.getMessage(),e);
        }
        finally {
            lock.readLock().unlock();
        }
        return null;
    }

    @Override
    public boolean addAll(Collection<? extends TaskItem<T>> items) {
        if(CollectionUtils.isEmpty(items)){
            return false;
        }
        return super.addAll(items);
    }

    @Override
    public int size() {
        return awaitQueueSize.get() + failedQueueSize.get();
    }

    @Override
    public boolean offer(TaskItem<T> item) {
        try {
            lock.writeLock().lockInterruptibly();
            item.setStatus(TaskStatus.READY);
            awaitQueueSize.incrementAndGet();
            return awaitQueue.add(item);
        }
        catch (Exception e){
            log.error(TRY_LOCK_TIMEOUT+e.getMessage(),e);
        }
        finally {
            lock.writeLock().unlock();
        }
        return false;
    }

    /**
     * 优先从等待队列获取，其次从失败队列获取
     * @return
     */
    @Override
    public TaskItem<T> poll() {
        try {
            lock.writeLock().lockInterruptibly();
            TaskItem<T> item = awaitQueue.poll();
            if (item == null) {
                item = failedQueue.poll();
                if(item != null){
                    failedQueueSize.decrementAndGet();
                }
            }
            else {
                awaitQueueSize.decrementAndGet();
            }
            if (item != null) {
                // 分发自动转为处理中
                item.setStatus(TaskStatus.PROCESSING);
                processItemMap.put(item.getId(),item);
                processItemSize.incrementAndGet();
                return item;
            }
        }
        catch (Exception e){
            log.error(TRY_LOCK_TIMEOUT+e.getMessage(),e);
        }
        finally {
            lock.writeLock().unlock();
        }
        return null;
    }

    @Override
    public TaskItem<T> peek() {
        try {
            lock.readLock().lockInterruptibly();
            TaskItem<T> item = awaitQueue.peek();
            if (item == null) {
                item = failedQueue.peek();
            }
            if (item != null) {
                return item;
            }
        }
        catch (Exception e){
            log.error(TRY_LOCK_TIMEOUT+e.getMessage(),e);
        }
        finally {
            lock.readLock().unlock();
        }
        return null;
    }

    @Override
    public void clear() {
        super.clear();
        awaitQueue.clear();
        failedQueue.clear();
        commitFailedQueue.clear();
        processItemMap.clear();
    }

    /**
     * 提交本地状态
     * @param id
     * @param status
     */
    public void commitStatus(long id, TaskProcessStatus status){
        TaskItem<T> taskItem = processItemMap.get(id);
        if(taskItem == null){
            return;
        }
        try{
            lock.writeLock().lockInterruptibly();
            switch (status){
                case SUCCESS:{
                    taskItem.setStatus(TaskStatus.SUCCESS);
                    break;
                }
                case COMMIT_FAILED:{
                    int failedCount = taskItem.getFailedCount();
                    if(failedCount <= MAX_FAILED_RETRY_COUNT){
                        taskItem.setFailedCount(failedCount +1);
                        commitFailedQueue.add(taskItem);
                    }
                    else {
                        taskItem.setStatus(TaskStatus.COMMIT_FAILURE);
                    }
                    break;
                }
                case FAILED:{
                    int failedCount = taskItem.getFailedCount();
                    if(failedCount <= MAX_FAILED_RETRY_COUNT){
                        taskItem.setFailedCount(failedCount +1);
                        failedQueue.add(taskItem);
                        failedQueueSize.incrementAndGet();
                    }
                    else {
                        taskItem.setStatus(TaskStatus.FAILURE);
                    }
                    break;
                }
                case ERROR:{
                    taskItem.setStatus(TaskStatus.FAILURE);
                    break;
                }
                default:
                    break;
            }
            processItemMap.remove(id);
            processItemSize.decrementAndGet();
        }
        catch (Exception e){
            log.error(TRY_LOCK_TIMEOUT+e.getMessage(),e);
        }
        finally {
            lock.writeLock().unlock();
        }
    }

    public void printSize(){
        StringBuffer sbf = new StringBuffer();
        sbf.append("awaitQueue:").append(awaitQueue.size()).append(";")
                .append("awaitQueueSize:").append(awaitQueueSize).append(";")
                .append("processItemMap:").append(processItemMap.size()).append(";")
                .append("processItemSize:").append(processItemSize.get()).append(";")
                .append("failedQueue:").append(failedQueue.size()).append(";")
                .append("failedQueueSize:").append(failedQueueSize).append(";")
                .append("commitFailedQueue:").append(commitFailedQueue.size()).append(";");
        log.debug(sbf.toString());
    }
}
